package org.poem.kafka;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.poem.ExecTaskDetailPlanVO;
import org.poem.canal.CanalHandleEntry;
import org.poem.exec.handler.ClientHandlerScheduledExecutor;
import org.poem.exec.register.CanalHandlerRegister;
import org.poem.exec.register.TaskRegisterFilter;
import org.poem.enums.KafkaEnums;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.*;

@Component
public class SyncKafkaConsumer implements CommandLineRunner {

    public static final Logger logger = LoggerFactory.getLogger(SyncKafkaConsumer.class);
    private static Consumer<String, String> consumer;

    private static List<CanalHandleEntry> canalHandleEntries = Lists.newArrayList();

    @Value("${kafka.bootstrap-servers.address}")
    private String address;

    private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

    private boolean close = false;

    /**
     * 启动就可以执行
     */
    @PostConstruct
    public void init() {
        Properties props = new Properties();
        props.put("bootstrap.servers", address);
        props.put("group.name", "CountryCounter"); // 当前group组中的名字
        // （在相同的group组中做consumer的qufe
        props.put("group.id", "CountryCounter");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(props);

        //注册JVM关闭时的回调钩子，当JVM关闭时调用此钩子。
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                close = true;
                System.out.println("Starting exit...");
                //调用消费者的wakeup方法通知主线程退出
                consumer.close();
            }
        });
    }


    /**
     * 发送消息
     */
    public void consumer() {
        try {
            consumer.subscribe(Collections.singletonList(KafkaEnums.DATA_FULL_TRANSFORM_TO_PRODUCER_TOPIC_MYSQL), new HandleRebalance());
            while (true) {
                if (!close) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMinutes(5000));
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
                        currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, "no metadata"));
                        String value = record.value();
                        ExecTaskDetailPlanVO execTaskDetailPlanVO = JSONObject.parseObject(value, ExecTaskDetailPlanVO.class);
                        ClientHandlerScheduledExecutor.handlerFullConsumer(canalHandleEntries, execTaskDetailPlanVO);
                    }
                    consumer.commitAsync(currentOffsets, null);
                } else {
                    break;
                }

            }
        } catch (WakeupException e) {
            // ignore, we're closing
        } catch (Exception e) {
            logger.error("Unexpected error", e);
        } finally {
            try {
                consumer.commitSync(currentOffsets);
            } finally {
                consumer.close();
                System.out.println("Closed consumer and we are done");
            }
        }
    }

    @Override
    public void run(String... args) throws Exception {
        canalHandleEntries = CanalHandlerRegister.getHandler(new TaskRegisterFilter() {
            @Override
            public Boolean accept(CanalHandleEntry zgTask) {
                return true;
            }
        }, "org.poem");
        this.consumer();
    }

    /**
     * rebalance
     */
    private class HandleRebalance implements ConsumerRebalanceListener {
        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        }

        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            System.out.println("Lost partitions in rebalance.Committing current offsets:" + currentOffsets);
            consumer.commitSync(currentOffsets);
        }
    }
}

